package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * Author: Felix
 * Date: 2022/5/11
 * Desc: 交易域取消订单事务事实表
 */
public class DwdTradeOrderCancelDetail {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2. 检查点相关设置(略)

        // TODO 3. 读取 Kafka dwd_trade_order_pre_process 主题数据
        tableEnv.executeSql("" +
                "create table dwd_trade_order_pre_process(  " +
                "id string, " +
                "order_id string, " +
                "course_id string, " +
                "course_name string, " +
                "date_id string, " +
                "create_time string, " +
                "original_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "od_ts string, " +
                "user_id string, " +
                "province_id string, " +
                "operate_date_id string, " +
                "create_time0 string, " +
                "update_time string, " +
                "order_status string, " +
                "`type` string, " +
                "`old` map<string,string>, " +
                "oi_ts string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + MyKafkaUtil.getKafkaDDL(
                "dwd_trade_order_pre_process", "dwd_trade_cancel_detail"));

        // TODO 4. 筛选取消订单明细数据
        Table filteredTable = tableEnv.sqlQuery("" +
                "select " +
                "id, " +
                "order_id, " +
                "user_id, " +
                "course_id, " +
                "course_name, " +
                "province_id, " +
                "date_id, " +
                "create_time, " +
                "original_amount, " +
                "coupon_reduce, " +
                "final_amount, " +
                "od_ts ts, " +
                "row_op_ts " +
                "from dwd_trade_order_pre_process  " +
                "where `type` = 'update'  " +
                "and `old`['order_status'] is not null  " +
                "and order_status = '1003'");
        tableEnv.createTemporaryView("filtered_table", filteredTable);

        // TODO 5. 建立 Upsert-Kafka dwd_trade_cancel_detail 表
        tableEnv.executeSql("create table dwd_trade_cancel_detail(  " +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "province_id string, " +
                "date_id string, " +
                "create_time string, " +
                "original_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cancel_detail"));

        // TODO 6. 将数据写出到 Kafka
        tableEnv.executeSql(
                "insert into dwd_trade_cancel_detail select * from filtered_table");
    }

}
